"""
Case Type   : Audit
Case Name   : 存储过程审计合理报错
Create At   : 2025/03/31
Owner       : liu-tong-8848
Description :
              1.审计检查DML关键字拼写错误文本
              sys.argv[1] = $GAUSSHOME/lib/postgresql/libog_query.so
              sys.argv[2] = "create or replace procedure test_audit_to_table(list in varchar2)
                  as
                  tablen binary_integer := 0;
                  tab varchar2[];
                  i int;
                  begin
                      for i in 1..tablen loop
                  end loop;
                  end;
                  /"
              # -*- coding: utf-8 -*-
              import sys
              import ctypes
              class OgQueryParseResult(ctypes.Structure):
                  _fields_ = [
                      ("parse_tree_json", ctypes.c_char_p),
                      ("err_msg", ctypes.c_char_p),
                      ("is_passed", ctypes.c_bool)
                  ]
              libpg_query = ctypes.CDLL(sys.argv[1])
              libpg_query.raw_parser_opengauss_dolphin.restype = OgQueryParseResult
              libpg_query.raw_parser_opengauss_dolphin.argtypes = [ctypes.c_char_p]
              sql_bytes = sys.argv[2].encode('utf-8')
              result = libpg_query.raw_parser_opengauss_dolphin(sql_bytes)
              print(sql_bytes)
              print(result.parse_tree_json)
              print(result.err_msg)
              print(result.is_passed)
Expect      :
              1.失败，合理报错
History     :
"""

import unittest

from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class Audit(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.constant = Constant()
        self.log.info('==Opengauss_Function_Grammar_Audit_Case003开始==')
        self.python_file_name = "test_audit.py"
        self.db_user_node = Node(node='PrimaryDbUser')

    def test_add_dimension(self):
        text = '----step1: 审计存储过程 expect: 失败，合理报错----'
        self.log.info(text)
        init_order_1 = "echo '" \
                       "# -*- coding: utf-8 -*-\n" \
                       "import sys\n" \
                       "import ctypes\n" \
                       "\n" \
                       "class OgQueryParseResult(ctypes.Structure):\n" \
                       "    _fields_ = [\n" \
                       "        (\"parse_tree_json\", ctypes.c_char_p),\n" \
                       "        (\"err_msg\", ctypes.c_char_p),\n" \
                       "        (\"is_passed\", ctypes.c_bool)\n" \
                       "    ]\n" \
                       "\n" \
                       "' > " + self.python_file_name
        init_order_2 = "echo \"" \
                       "libpg_query = ctypes.CDLL(sys.argv[1])\n" \
                       "libpg_query.raw_parser_opengauss_dolphin.restype = OgQueryParseResult\n" \
                       "libpg_query.raw_parser_opengauss_dolphin.argtypes = [ctypes.c_char_p]\n" \
                       "\n" \
                       "sql_bytes = sys.argv[2].encode('utf-8')\n" \
                       "result = libpg_query.raw_parser_opengauss_dolphin(sql_bytes)\n" \
                       "print(sql_bytes)\n" \
                       "print(result.parse_tree_json)\n" \
                       "print(result.err_msg)\n" \
                       "print(result.is_passed)\n" \
                       "\" >> " + self.python_file_name
        self.log.info(init_order_1)
        self.log.info(init_order_2)
        init_result_1 = self.db_user_node.sh(init_order_1).result()
        init_result_2 = self.db_user_node.sh(init_order_2).result()
        self.assertNotIn('ERROR', init_result_1, "执行失败")
        self.assertNotIn('ERROR', init_result_2, "执行失败")

        self.log.info("cat " + self.python_file_name)
        cat_result = self.db_user_node.sh("cat " + self.python_file_name).result()
        self.log.info(cat_result)

        test_so_file_name = "$GAUSSHOME/lib/postgresql/libog_query.so"
        test_sql = "create or replace procedure test_audit_to_table(list in varchar2) " \
                   + "as " \
                   + "tablen binary_integer := 0; " \
                   + "tab varchar2[]; " \
                   + "i int; " \
                   + "begin " \
                   + "for i in 1..tablen loop " \
                   + "end loop; " \
                   + "end; " \
                   + "/"
        test_order = "python3 " + self.python_file_name + " " + test_so_file_name + " \"" + test_sql + "\""
        self.log.info(test_order)
        source_order = "source " + macro.DB_ENV_PATH + ";"
        self.log.info(source_order)
        test_result = self.db_user_node.sh(source_order + test_order).result()
        self.log.info(test_result)
        self.assertIn('syntax error at or near', test_result.splitlines()[-2], "执行失败")
        self.assertIn('False', test_result.splitlines()[-1], "执行失败")

    def tearDown(self):
        text = '----step2: 清理环境 expect: 成功----'
        self.log.info(text)
        drop_python_order = "rm -rf " + self.python_file_name
        drop_python_result = self.db_user_node.sh(drop_python_order).result()
        self.log.info(drop_python_result)
        self.assertNotIn('ERROR', drop_python_result, "执行失败")
        self.log.info('==Opengauss_Function_Grammar_Audit_Case003完成==')
